package com.example.udp.server;

import com.example.node.config.NodeConfig;
import com.example.udp.message.UdpMessage;
import com.example.udp.utils.JavaSerializableUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class UdpServer {

    Logger logger = LoggerFactory.getLogger(getClass());

    private String ip = "255.255.255.255";

    private int port = 9999;

    private int period = 10;

    private int core = 10;

    private NodeConfig nodeConfig;

    private InetAddress inetAddress;

    private DatagramSocket datagramSocket;

    private ScheduledExecutorService scheduledExecutorService;

    private UdpServerMessageListener udpServerMessageListener;

    public UdpServer(NodeConfig nodeConfig, UdpServerMessageListener udpServerMessageListener) {
        this.nodeConfig = nodeConfig;
        this.udpServerMessageListener = udpServerMessageListener;
    }

    public UdpServer(String ip, int port, int period, int core, NodeConfig nodeConfig, UdpServerMessageListener udpServerMessageListener) {
        this.ip = ip;
        this.port = port;
        this.period = period;
        this.core = core;
        this.nodeConfig = nodeConfig;
        this.udpServerMessageListener = udpServerMessageListener;
    }

    public void start() throws Exception {
        this.inetAddress = InetAddress.getByName(ip);
        this.datagramSocket = new DatagramSocket();
        this.datagramSocket.setBroadcast(true);
        this.scheduledExecutorService = new ScheduledThreadPoolExecutor(core);
        this.scheduledExecutorService.scheduleAtFixedRate(new UdpPingThread(), 0, period, TimeUnit.SECONDS);
        this.scheduledExecutorService.execute(new Runnable() {
            @Override
            public void run() {
                MulticastSocket multicastSocket = null;
                try {
                    multicastSocket = new MulticastSocket(port);
                    multicastSocket.joinGroup(inetAddress);
                } catch (Exception e) {
                    logger.warn("广播消息接收器启动失败: " + e.getMessage());
                }
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        byte[] buffer = new byte[1024];
                        DatagramPacket datagramPacket = new DatagramPacket(buffer, buffer.length);
                        multicastSocket.receive(datagramPacket);
                        if (datagramPacket.getLength() > 0) {
                            UdpMessage udpMessage = (UdpMessage) JavaSerializableUtils.deSerializable(buffer);
                            scheduledExecutorService.execute(new UdpServerMessageHandler(udpMessage));
                        }
                    } catch (Exception e) {
                        if (multicastSocket != null) {
                            multicastSocket.close();
                        }
                        if (e instanceof InterruptedException) {
                            Thread.currentThread().interrupt();
                        }
                    }
                }
            }
        });
    }

    public void stop() {
        if (this.datagramSocket != null) {
            this.datagramSocket.close();
        }
        this.scheduledExecutorService.shutdownNow();
    }

    class UdpPingThread implements Runnable {

        @Override
        public void run() {
            if (datagramSocket != null) {
                try {
                    UdpMessage udpMessage = new UdpMessage(nodeConfig);
                    byte[] bytes = JavaSerializableUtils.serializable(udpMessage);
                    DatagramPacket datagramPacket = new DatagramPacket(bytes, bytes.length, inetAddress, port);
                    datagramSocket.send(datagramPacket);
                } catch (Exception e) {
                    logger.warn("udp server send message error: " + e.getMessage());
                }
            }
        }
    }

    class UdpServerMessageHandler implements Runnable {

        private UdpMessage udpMessage;

        public UdpServerMessageHandler(UdpMessage udpMessage) {
            this.udpMessage = udpMessage;
        }

        @Override
        public void run() {
            udpServerMessageListener.onMessage(udpMessage);
        }
    }

    public interface UdpServerMessageListener {

        public void onMessage(UdpMessage udpMessage);
    }
}

